package wwl.lsf.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MsgConsumer {

  @KafkaListener(topics = "hello",groupId = "ttkfz")
  public void consume(ConsumerRecord<String, Object> record,
      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer, Acknowledgment ack) {

    log.error("单独消费者消费消息,topic= {} ,content = {}",topic,record.value());
    log.error("consumer content = {}",consumer);
    ack.acknowledge();
  }
//
  @KafkaListener(topics = "hello3333")
  public void consume2(ConsumerRecord<String, Object> record,
      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer, Acknowledgment ack) {

    log.error("单独消费者消费消息,topic= {} ,content = {}",topic,record.value());
    log.error("consumer content = {}",consumer);
    ack.acknowledge();
  }

  @KafkaListener(topics = "hello")
  public void consume3(ConsumerRecord<String, Object> record,
      @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer, Acknowledgment ack) {

    log.error("单独消费者消费消息,topic= {} ,content = {}",topic,record.value());
    log.error("consumer content = {}",consumer);
    ack.acknowledge();
  }

}
